package com.chenjj.bigdata.flink.window.function

import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
  * 自定义ProcessWindowFunction实现指标统计，分别求每个key下的最小值，最大值，求和，平均值，窗口的start, 窗口end
  */
class MyProcessWindowFunction extends ProcessWindowFunction[(String,Long,Int),(String,Long,Long,Long,Long,Long,Long),String,TimeWindow]{
  override def process(key: String, context: Context, elements: Iterable[(String, Long, Int)], out: Collector[(String, Long, Long, Long, Long, Long, Long)]): Unit = {
    val min = elements.map(_._3).min
    val max = elements.map(_._3).max
    val sum = elements.map(_._3).sum
    val avg = sum / elements.size
    val start = context.window.getStart
    val end = context.window.getEnd

    out.collect((key,min,max,sum,avg,start,end))
  }
}
